package com.afdor.rws.dlock.zookeeper;

import com.afdor.rws.annotation.extension.Extension;
import com.afdor.rws.dlock.lock.ReentrantLock;
import com.afdor.rws.dlock.lock.ReentrantReadWriteLock;
import com.google.common.base.Predicate;
import lombok.EqualsAndHashCode;
import lombok.ToString;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.framework.recipes.locks.LockInternalsDriver;
import org.apache.curator.framework.recipes.locks.PredicateResults;
import org.apache.curator.framework.recipes.locks.StandardLockInternalsDriver;
import org.apache.curator.shaded.com.google.common.collect.ImmutableList;
import org.apache.curator.shaded.com.google.common.collect.Iterables;
import org.apache.zookeeper.KeeperException;

import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;

/**
 * Zookeeper实现的分布式可重入读写锁
 *
 * @author 悭梵
 * @date Created in 2018-08-31 17:45
 */
@Slf4j
@ToString
@EqualsAndHashCode
@Extension(value = "zookeeper", scope = Extension.Scope.PROTOTYPE, order = Extension.NORMAL_ORDER + 100, interfaceClass = ReentrantReadWriteLock.class)
public class DistributedReentrantReadWriteLock implements ReentrantReadWriteLock {

    private static final String READ_LOCK_NAME = "__READ__";
    private static final String WRITE_LOCK_NAME = "__WRIT__";
    private static final String ROOT_PATH = "/RWS/Distributed/ReentrantReadWriteLock/";

    private final ReentrantLock readMutex;
    private final ReentrantLock writeMutex;

    public DistributedReentrantReadWriteLock(CuratorFramework client, String lockId) {
        if (StringUtils.isBlank(lockId)) {
            throw new NullPointerException("锁标识为空");
        }
        if (client == null) {
            throw new NullPointerException("Curator客户端为空");
        }
        lockId = StringUtils.trim(lockId);
        String path = ROOT_PATH + lockId;
        this.writeMutex = new DistributedReentrantReadWriteLock.InternalDistributedReentrantLock(client, path, WRITE_LOCK_NAME, 1, new DistributedReentrantReadWriteLock.SortingLockInternalsDriver() {
            @Override
            public PredicateResults getsTheLock(CuratorFramework client, List<String> children, String sequenceNodeName, int maxLeases) throws Exception {
                return super.getsTheLock(client, children, sequenceNodeName, maxLeases);
            }
        });
        this.readMutex = new DistributedReentrantReadWriteLock.InternalDistributedReentrantLock(client, path, READ_LOCK_NAME, Integer.MAX_VALUE, new DistributedReentrantReadWriteLock.SortingLockInternalsDriver() {
            @Override
            public PredicateResults getsTheLock(CuratorFramework client, List<String> children, String sequenceNodeName, int maxLeases) throws Exception {
                return DistributedReentrantReadWriteLock.this.readLockPredicate(children, sequenceNodeName);
            }
        });
    }


    @Override
    public ReentrantLock readLock() {
        return this.readMutex;
    }

    @Override
    public ReentrantLock writeLock() {
        return this.writeMutex;
    }

    private PredicateResults readLockPredicate(List<String> children, String sequenceNodeName) throws Exception {
        if (((DistributedReentrantReadWriteLock.InternalDistributedReentrantLock) this.writeMutex).isOwnedByCurrentThread()) {
            return new PredicateResults((String) null, true);
        } else {
            int index = 0;
            int firstWriteIndex = 2147483647;
            int ourIndex = -1;

            String node;
            for (Iterator var6 = children.iterator(); var6.hasNext(); ++index) {
                node = (String) var6.next();
                if (node.contains(WRITE_LOCK_NAME)) {
                    firstWriteIndex = Math.min(index, firstWriteIndex);
                } else if (node.startsWith(sequenceNodeName)) {
                    ourIndex = index;
                    break;
                }
            }

            if (ourIndex < 0) {
                throw new KeeperException.NoNodeException("Sequential path not found: " + sequenceNodeName);
            }
            boolean getsTheLock = ourIndex < firstWriteIndex;
            node = getsTheLock ? null : (String) children.get(firstWriteIndex);
            return new PredicateResults(node, getsTheLock);
        }
    }

    @Slf4j
    static class InternalDistributedReentrantLock extends DistributedReentrantLock {

        private final String lockName;

        InternalDistributedReentrantLock(CuratorFramework client, String path, String lockName, int maxLeases, LockInternalsDriver driver) {
            this.path = path;
            this.client = client;
            this.lockName = lockName;

            try {
                Constructor<InterProcessMutex> constructor = InterProcessMutex.class.getDeclaredConstructor(CuratorFramework.class, String.class, String.class, int.class, LockInternalsDriver.class);
                constructor.setAccessible(true);
                interProcessMutex = constructor.newInstance(client, path, lockName, maxLeases, driver);
            } catch (InstantiationException | IllegalAccessException | InvocationTargetException | NoSuchMethodException e) {
                throw new RuntimeException("实例锁失败", e);
            }
        }

        @Override
        public void init(CuratorFramework client, String lockId) {
            throw new IllegalStateException("已通过构造方法初始化，无需再执行该初始化方法。");
        }

        protected boolean isOwnedByCurrentThread() {
            return interProcessMutex.isOwnedByCurrentThread();
        }

        public Collection<String> getParticipantNodes() throws Exception {
            Collection<String> nodes = interProcessMutex.getParticipantNodes();
            Iterable<String> filtered = Iterables.filter(nodes, new Predicate<String>() {
                @Override
                public boolean apply(String node) {
                    return node.contains(DistributedReentrantReadWriteLock.InternalDistributedReentrantLock.this.lockName);
                }
            });
            return ImmutableList.copyOf(filtered);
        }
    }


    private static class SortingLockInternalsDriver extends StandardLockInternalsDriver {
        private SortingLockInternalsDriver() {
        }

        @Override
        public final String fixForSorting(String str, String lockName) {
            str = super.fixForSorting(str, READ_LOCK_NAME);
            str = super.fixForSorting(str, WRITE_LOCK_NAME);
            return str;
        }
    }
}

